package com.pxene.dmp.mapper;

import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.Date;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import com.pxene.dmp.common.URLUtils;

public class GetWeixinBizMapper extends Mapper<LongWritable, Text, LongWritable, Text> {
	
	private URL urlParser;
	private LongWritable resultKey = new LongWritable();
	private Text resultVal = new Text();
	
	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		Configuration conf = context.getConfiguration();
		// 设置配置文件名
		System.setProperty("config-file", conf.get("config-file"));
	}

	@Override
	protected void map(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		String[] columns = value.toString().split("\\|");
		if (columns.length != 6) {
			return;
		}
		String imsi = columns[0];
		String bsid = columns[1];
		String starttime = columns[2];
		String url = columns[3];
		String province = columns[4];
		String dataType = columns[5];
		
		try {
			urlParser = new URL(url);
		} catch(MalformedURLException e) {
			return;
		}
		
		Map<String, String> map = URLUtils.getQueryMap(urlParser.getQuery());
		String biz = "";
		String mid = "";
		String idx = "";
		String sn = "";
		if (map.containsKey("__biz")) {
			biz = map.get("__biz").trim();
		}
		if (map.containsKey("mid")) {
			mid = map.get("mid").trim();
		}
		if (map.containsKey("idx")) {
			idx = map.get("idx").trim();
		}
		if (map.containsKey("sn")) {
			sn = map.get("sn").trim();
		}
		
		if (biz.length() > 0 && mid.length() > 0 && idx.length() > 0 && sn.length() > 0) {
			resultKey.set(new Date().getTime());
			resultVal.set(imsi + "|" + bsid + "|" + starttime + "|" + biz + ";" + mid + ";" + idx + ";" + sn + "|" + province + "|" + dataType);
			context.write(resultKey, resultVal);
		}
	}
	
}
